package indi.zhifa.study2025.test.kafka.consumer.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class KafkaConsumerListener {

    private Map<String,Long> mMsgIdx;

    public KafkaConsumerListener() {
        mMsgIdx = new ConcurrentHashMap<>();
    }

    @KafkaListener(topics = "my-test-topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record) {
        String key = record.key();           // 获取消息的 key
        String value = record.value();       // 获取消息的 value
        String topic = record.topic();       // 获取消息的 topic
        int partition = record.partition(); // 获取消息的分区
        long offset = record.offset();      // 获取消息的偏移量
        long timestamp = record.timestamp(); // 获取消息的时间戳

        // 处理消息（这里我们只是打印消息）
        System.out.println("Consumed record: ");
        System.out.println("Key: " + key);
        System.out.println("Value: " + value);
        System.out.println("Topic: " + topic);
        System.out.println("Partition: " + partition);
        System.out.println("Offset: " + offset);
        System.out.println("Timestamp: " + timestamp);

        if(StringUtils.hasText(key)){
           Long idx = mMsgIdx.get(key);
           if(idx == null){
               idx = 0l;
           }
           idx = idx + 1;
           mMsgIdx.put(key, idx);

           System.out.println(key+"的第"+idx+"个消息");
        }
    }
}
